package com.itqiqi.api.Transform;

import com.itqiqi.api.pojo.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

import java.util.Collections;


public class TransformTest5_RichFunction {

    public static void main(String[] args) throws Exception {

        // 获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        DataStreamSource<String> dataStream = env.readTextFile("input/sensor.txt");

        // 转换成SensorReading类型
        SingleOutputStreamOperator<SensorReading> mapStream = dataStream.map((MapFunction<String, SensorReading>) s -> {
            String[] fields = s.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        // 转换成id和分区标号的二元组
        SingleOutputStreamOperator<Tuple2<String, Integer>> resStream = mapStream.map(new RichMapFunction<SensorReading, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(SensorReading sensorReading) throws Exception {
                return new Tuple2<>(sensorReading.getId(), getRuntimeContext().getIndexOfThisSubtask());
            }

            @Override
            public void open(Configuration parameters) throws Exception {
                // 以下可以做一些初始化工作，例如建立一个和 HDFS 的连接
                System.out.println("open");
            }

            @Override
            public void close() throws Exception {
                // 以下做一些清理工作，例如断开和 HDFS 的连接
                System.out.println("close");
            }
        });

        resStream.print("RichMapFunction");
        env.execute();

    }
}
